package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;
import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 对内存中的所有数据进行管理
 */
public class MemoryDataCenter {
    //exchangeName - Exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //queueName - MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //exchangeName - queueName
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //messageId - Message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //queueName - message对象
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 第一个 key 是 queueName, 第二个 key 是 messageId
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    /**
     * 添加交换机
     * @param exchange
     */
    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功! exchangeName=" + exchange.getName());
    }

    /**
     * 获取交换机
     * @param exchange
     * @return
     */
    public Exchange getExchange(String exchange){
        return exchangeMap.get(exchange);
    }

    /**
     * 删除交换机
     * @param exchangeName
     */
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName=" + exchangeName);
    }

    /**
     * 添加队列
     * @param queue
     */
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功! queueName=" + queue.getName());
    }

    /**
     * 获取队列
     * @param queueName
     * @return
     */
    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    /**
     * 删除队列
     * @param queueName
     */
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功! queueName=" + queueName);
    }

    /**
     * 新建绑定
     * @param binding
     * @throws MqException
     */
    public void insertBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());

        //不为空则抛出异常 为空则插入
        synchronized (bindingMap) {
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter]绑定已经存在！exchangeName" + binding.getExchangeName() +
                        ",queueName" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功! exchangeName=" + binding.getExchangeName()
                + ", queueName=" + binding.getQueueName());
    }


    /**
     * 获得绑定
     * @param exchangeName
     * @param queueName
     * @return
     */
    // 1. 根据exchangeName和queueName获得唯一的绑定
    public Binding getBinding(String exchangeName,String queueName){

        ConcurrentHashMap<String,Binding> bindingMap =bindingsMap.get(exchangeName);
        if(bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
        
    }
    // 2. 根据exchangeName获得所有的绑定
    public ConcurrentHashMap<String ,Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    /**
     * 删除绑定
     * @param binding
     * @throws MqException
     */
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 该交换机没有绑定任何队列. 报错.
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName()
                    + ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功! exchangeName=" + binding.getExchangeName()
                + ", queueName=" + binding.getQueueName());
    }

    /**
     * 添加消息
     * @param message
     */
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter]消息添加成功，messageId="+message.getMessageId());
    }

    /**
     * 根据消息id获取消息
     * @param messageId
     * @return
     */
    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

    /**
     * 根据消息id删除消息
     * @param messageId
     */
    public void removeMessage(String messageId){
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter]消息移除成功，messageId="+messageId);
    }


    /**
     * 发送消息到指定的队列
     * @param message
     * @param queue
     */
    public void sendMessage(Message message ,MSGQueue queue){
         //把消息放入到指定的队列数据结构中
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(),
                k->new LinkedList<>());
        synchronized (messages){
            messages.add(message);
        }
        //添加message
        addMessage(message);
        System.out.println("[MemoryDataCenter]消息被投递到队列中，messageId="+message.getMessageId());

    }

    /**
     * 从队列中取消息
     * @param queueName 队列名
     * @return
     */
    public Message pollMessage(String queueName) {
        // 根据队列名, 查找一下, 对应的队列的消息链表.
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            // 如果没找到, 说明队列中没有任何消息.
            if (messages.size() == 0) {
                return null;
            }
            // 链表中有元素, 就进行头删.
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出! messageId=" + currentMessage.getMessageId());
            return currentMessage;
        }
    }


    /**
     * 获取队列中的消息总数
     * @param queueName
     * @return
     */
    public int getMessageCount(String queueName){
        //根据queueName获得messages消息队列
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages ==null){
            return 0;
        }
        return messages.size();

    }

    /**
     * 添加未确认的消息
     * @param queueName 要添加的队列名
     * @param message 消息体
     */
    public void addMessageWaitAck(String queueName,Message message){
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k->new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列 messageId="+message.getMessageId());
    }


    /**
     * 删除未确认的消息（确认消息）
     * @param queueName
     * @param messageId
     */
    public void removeMessageWaitAck(String queueName,String messageId){
         ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
         if(messageHashMap == null){
             return;
         }
         messageHashMap.remove(messageId);
        System.out.println("[MemooryDataCenter] 消息已确认 messageId="+messageId);
    }


    /**
     * 获得指定的未确认的消息
     * @param queueName
     * @param messageId
     * @return
     */
    public Message getMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if(messageHashMap == null){
            return null;
        }
        return messageHashMap.get(messageId);

    }

//
//    // 这个方法就是从硬盘上读取数据, 把硬盘中之前持久化存储的各个维度的数据都恢复到内存中.
//    public void recovery1(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
//        // 0. 清空之前的所有数据
//        exchangeMap.clear();
//        queueMap.clear();
//        bindingsMap.clear();
//        messageMap.clear();
//        queueMessageMap.clear();
//        // 1. 恢复所有的交换机数据
//        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
//        for (Exchange exchange : exchanges) {
//            exchangeMap.put(exchange.getName(), exchange);
//        }
//        // 2. 恢复所有的队列数据
//        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
//        for (MSGQueue queue : queues) {
//            queueMap.put(queue.getName(), queue);
//        }
//        // 3. 恢复所有的绑定数据
//        List<Binding> bindings = diskDataCenter.selectAllBindings();
//        for (Binding binding : bindings) {
//            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
//                    k -> new ConcurrentHashMap<>());
//            bindingMap.put(binding.getQueueName(), binding);
//        }
//        // 4. 恢复所有的消息数据
//        //    遍历所有的队列, 根据每个队列的名字, 获取到所有的消息.
//        for (MSGQueue queue : queues) {
//            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
//            queueMessageMap.put(queue.getName(), messages);
//            for (Message message : messages) {
//                messageMap.put(message.getMessageId(), message);
//            }
//        }
//        // 注意!! 针对 "未确认的消息" 这部分内存中的数据, 不需要从硬盘恢复. 之前考虑硬盘存储的时候, 也没设定这一块.
//        // 一旦在等待 ack 的过程中, 服务器重启了, 此时这些 "未被确认的消息", 就恢复成 "未被取走的消息" .
//        // 这个消息在硬盘上存储的时候, 就是当做 "未被取走"

    /**
     * 从硬盘中恢复消息
     * @param diskDataCenter 硬盘数据
     * @throws IOException 
     * @throws MqException
     * @throws ClassNotFoundException
     */
        public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
            //清空内存中的数据
            exchangeMap.clear();
            queueMap.clear();
            bindingsMap.clear();
            queueMessageMap.clear();
            messageMap.clear();


            //1.恢复所有的交换机数据
            //获取到数据
            List<Exchange> exchanges = diskDataCenter.selectAllExchanges() ;
            for(Exchange exchange : exchanges){
                //放入map
                exchangeMap.put(exchange.getName(),exchange);
            }

            //2.恢复所有的队列数据
            List<MSGQueue> queues = diskDataCenter.selectAllQueues();
            for(MSGQueue queue : queues){
                queueMap.put(queue.getName(),queue);
            }

            //3.恢复所有的绑定数据
            List<Binding> bindings = diskDataCenter.selectAllBindings();
            for(Binding binding : bindings ){
                ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                        k-> new ConcurrentHashMap<>());
                bindingMap.put(binding.getQueueName(),binding);
            }

            //4.恢复所有的消息数据
            //遍历每个队列 从队列中获取消息
            for(MSGQueue queue : queues){
                LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
                //将队列存储起来
                queueMessageMap.put(queue.getName(),messages);
                //在这个队列下存储消息
                for(Message message : messages){
                    messageMap.put(message.getMessageId(),message);
                }
            }

        }

}
